package mapreduce.phase1;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.TreeSet;

import mapreduce.ColumnGrp;
import mapreduce.DataToPass;
import mapreduce.Point;
import mapreduce.UnionFind;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

/**
 * Localized run on points of the column to find connected components 
 *  -1 is key for vertices in column and -2 for edges, whole lower vertex is
 *  in column.
 *  Data Structures: Planar Sweep Algorithm
 */


@SuppressWarnings("deprecation")
public class Reduce1 extends MapReduceBase implements
Reducer<IntWritable, Text, IntWritable, Text> {

	public void reduce(IntWritable key, Iterator<Text> values,
			OutputCollector<IntWritable, Text> output, Reporter reporter)
	throws IOException {

		int group = key.get();
		List<Point> points = new ArrayList<Point>(DataToPass.DEFAULT_ELE_SIZE);
		
		UnionFind uf = new UnionFind();
		
		while (values.hasNext()) {
			Text t = values.next();
			Point p = Point.fromString(t.toString());
			points.add(p);
			uf.makeSet(p);
		}
		
		Collections.sort(points, new HComparator());
		int leftPos = 0;
		
		TreeSet<Point> bandPoints = new TreeSet<Point>(new VComparator());
		
		int sumVertex = 0;
		int sumEdge = 0;
		
		Iterator<Point> iterator = points.iterator();
		while(iterator.hasNext()) {
			Point p = iterator.next();
			double startX = p.getx(); 
			while(startX - points.get(leftPos).getx() > DataToPass.lmax) {
				bandPoints.remove(points.get(leftPos));
				leftPos++;
			}
			
			if(ColumnGrp.inActualBoundary(p, group))
				sumVertex++;
			
			//We take only the interesting candidates in the y axis
			double startY =  Math.max(p.gety()-DataToPass.lmax, 0.0);
			Point lowPoint = new Point(startX, startY);
			
			double endY = Math.min(p.gety()+DataToPass.lmax, 1.0);
			Point highPoint = new Point(startX, endY);
			
	        for (Point presentP : bandPoints.subSet(lowPoint, highPoint)) {
				if(p.distanceFrom(presentP) < DataToPass.lmax) {
					uf.union(p, presentP);
					if(ColumnGrp.inActualBoundary(presentP, group))
						sumEdge++;
					else if(presentP.getx() < ColumnGrp.getActualLeftBoundary(group) && p.getx()> ColumnGrp.getRightBoundryOfGroup(group-1))
						sumEdge++;
				}
	        }
	        bandPoints.add(p);
		}
		
		for (Point p : points) {
			Text t = new Text(p + " " + uf.find(p));
			output.collect(key, t);
		}

		key.set(-1);
		Text t = new Text(sumVertex + "");
		output.collect(key, t);
		
		key.set(-2);
		t.set(sumEdge + "");
		output.collect(key, t);
	}
}

class VComparator implements Comparator<Point> {
    @Override
    public int compare(Point p1, Point p2) {
        if (p1.gety() < p2.gety()) {
            return -1;
        }
        else if (p1.gety() > p2.gety()) {
            return 1;
        }
        else if (p1.getx() < p2.getx()) {
            return -1;
        }
        else if (p1.getx() > p2.getx()) {
            return 1;
        }
        return 0;
    }
};
 
class HComparator implements Comparator<Point> {
    @Override
    public int compare(Point p1, Point p2) {
        if (p1.getx() < p2.getx()) {
            return -1;
        }
        else if (p1.getx() > p2.getx()) {
            return 1;
        }
        else if (p1.gety() < p2.gety()) {
            return -1;
        }
        else if (p1.gety() > p2.gety()) {
            return 1;
        }
        return 0;
    }
};

